package lib ;

import java.util.Map ;

import com.fasterxml.jackson.databind.ObjectMapper;

import okhttp3.OkHttpClient ;
import okhttp3.Request ;
import okhttp3.RequestBody ;
import okhttp3.Response ;

import com.google.protobuf.Descriptors.Descriptor;
import com.google.protobuf.DynamicMessage;
import com.squareup.wire.schema.Location;
import com.squareup.wire.schema.internal.parser.ProtoParser;
import com.squareup.wire.schema.internal.parser.ProtoFileElement;
import protobuf.Lib;

import org.apache.zookeeper.ZooKeeper;


public class ApiClient {
    public static String invokeMetaServer(String apiServerNodes, String path) throws Exception {
        String envUrl = "http://" + apiServerNodes.split(",")[0] + path ;
        OkHttpClient http = new OkHttpClient() ;
        // MediaType SR_TYPE = MediaType.get("application/vnd.schemaregistry.v1+json");
        Request request = new Request.Builder().url(envUrl).build() ;

        try (Response response = http.newCall(request).execute() ) {
            return response.body().string();
        }
    }
    public static Map<String, String> getMetaEnv(String apiServerNodes) throws Exception {
        String response_str = invokeMetaServer(apiServerNodes, "/meta/raw_schema/env") ;
        Map<String, Object> json = new ObjectMapper().readValue(response_str, Map.class) ;
        return (Map<String, String>)json.get("message") ;
    }
    public static String getMetaSchema(String apiServerNodes) throws Exception {
        return invokeMetaServer(apiServerNodes, "/meta/avsc_schema_simple/avdl.data_buffer:dcSdkPlatformBulk") ;
    }
    public static String getMetaCommonSchema(String apiServerNodes) throws Exception {
        return invokeMetaServer(apiServerNodes, "/meta/avsc_schema_simple/avdl.data_buffer:dcSdkCommonBulk") ;
    }
    public static String getMetaGenericSchema(String apiServerNodes) throws Exception {
        return invokeMetaServer(apiServerNodes, "/meta/avsc_schema_simple/avdl.data_buffer:dcSdkGenericBulk") ;
    }

    public static Descriptor getHdfsDescriptor() {
        String schema_s = "syntax = \"proto3\";\n"
                        + "message ActiveNodeInfo {\n"
                        + "  string nameserviceId = 1;\n"
                        + "  string namenodeId = 2;\n"
                        + "  string hostname = 3;\n"
                        + "  int32 port = 4;\n"
                        + "  int32 zkfcPort = 5;\n"
                        + "}\n";
        String message_name = "ActiveNodeInfo";

        ProtoFileElement proto = ProtoParser.Companion.parse(Location.get(""), schema_s);
        return Lib.toDynamicSchema("default", proto, null).getMessageDescriptor(message_name);
                                
    }
    public static DynamicMessage getHdfsMaster(ZooKeeper zk, Descriptor descriptor, String hdfsHa) throws Exception {
        String zkNode = String.format("/hadoop-ha/%s/ActiveStandbyElectorLock", hdfsHa) ;
        return DynamicMessage.parseFrom(descriptor, zk.getData(zkNode, false, null)) ;
    }
}
